{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 使用appnium和路路通移动版获取时刻表  \n",
    "<b>实际上在开发中appnium已经不在维护且配置较为困难,仅离线使用</b>    \n",
    "下次会退回重新使用selenium  \n",
    "\n",
    "## 当前状态  \n",
    "- 主要由于路路通相较于12306在时刻表层面提供更多的信息  \n",
    "    - 如停到站时间及始发终到等信息丰富,车站大屏由于网络问题不好使用  \n",
    "    - 而路路通网页版不提供此类信息(12306不太好作),难以使用web自动化爬取  \n",
    "    - 这里使用appnium和模拟器来获得数据,但是配置较为复杂  \n",
    "    - 检票口(映射股道)需要依照车次再次查询如网页版12306或是这里的车次查询\n",
    "\n",
    "- 目前爬取效果,处理等和前面武汉枢纽相同\n",
    "    - 基本实现自动化,但是由于车站和车次页面不同,需要手动切换分辨率,也可以开始时直接拉满防止画面溢出,目前暂时不知道更好的方法\n",
    "    - 目前重点为完善车次爬取部分改为获取停站的上一站和下一站,可以解决进路冲突的问题如 济南东→济南西 和 济南→济南西 均为上行导致无法区分并获取更详细的信息\n",
    "    - 完善度表格:    <br>\n",
    "\n",
    "|组件信息|完善程度|当前状态|日后计划|  \n",
    "|---|---|---|---| \n",
    "车站|整体较为成熟完善|可以按照单一地区车站模糊匹配,精确匹配后续筛选|暂无\n",
    "车次|信息较为完善但是观感不佳|可以获得单一车次给出的大部分信息如停站检票口等.<br>主要为从车次查询-检票口查询获得|考虑压缩为字典样式以更好的存储信息\n",
    "检票口/股道 | 不完善，需要二次处理|如不需要车站爬取可以直接由单个车站获得足够信息,<br>无需经过查询车次继续处理。若需要检票口信息需要在获取完成车站的车次信息后再查询检票口|完善筛选逻辑\n",
    "立折车辆 | 纯手动，没有处理|手动查找修改|计划在日后由车次部分下交路表完成|计划在车次查询部分加上交路表部分信息获取\n",
    "越行车辆 |纯手动，完全没有处理|没有计划|不知道怎么获知越行车信息\n",
    "  \n",
    "## 使用的库函数信息和达成的基本效果\n",
    "<b>主要为了提高复用性功能大体分为了三个部分，数据处理可以单独由excel或csv使用无需使用爬虫</b>  \n",
    "- 获取整体信息-自动化爬虫部分\n",
    "    - appnium的APP自动化。<b>实际上在开发中appnium已经不在维护,仅离线使用</b>  \n",
    "    - 一般的io文本处理部分，输出为csv样式的txt文本文件\n",
    "- 初步信息处理-数据处理/数据分析\n",
    "    - 整理出较为规整的数据，主要按照车次来汇总数据便于后续操作\n",
    "    - 输出符合游戏地图要求的时刻表信息或是进行其他分析  \n",
    "- 没啥用的可视化\n",
    "    - 主要是pyecharts，输出一些可交互的图片更为直观\n",
    "\n",
    "----\n",
    "## 第一部分，定义使用的库函数和常量信息\n",
    "### 内容\n",
    "库函数主要分为三大部分：获取数据--处理数据--<del>分析（可视化）数据</del>。即下面引用部分大体分为了三部分  \n",
    "而常量部分主要用于处理数据生成符合游戏格式的时刻表，这部分会随着函数的完善而更为精简  \n",
    "以及最后的模拟器信息用于appnium连接模拟器获取路路通信息  \n",
    "### 待完善/优化的部分\n",
    "1. 编组映射待交路表信息获取完成后会按照车型映射如长编和重连，即使用下面marshalling字典\n",
    "2. 速度等级也会按照车型信息重新映射\n",
    "3. 建立进场路径映射相关常量来完善分场式车站可能的调度问题，提高调度自动化程度\n",
    "4. ....."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import appium  # app自动化使用，获取信息\n",
    "from appium import webdriver\n",
    "from appium.webdriver.webdriver import WebDriver\n",
    "from appium.webdriver.common.appiumby import AppiumBy\n",
    "from appium.webdriver.extensions.android.nativekey import AndroidKey\n",
    "from appium.webdriver.common.touch_action import TouchAction\n",
    "import time\n",
    "\n",
    "import pandas\n",
    "\n",
    "# 数据分析/数据可视化使用\n",
    "import matplotlib\n",
    "import pyecharts\n",
    "import plotly.figure_factory as ff\n",
    "from pyecharts import options as opts\n",
    "from pyecharts.charts import Bar, Timeline\n",
    "\n",
    "\n",
    "trainType = [\"普客\", \"新空调快速\", \"新空调特快\", \"新空调直快\", \"动车\", \"高速\", \"城际\"]\n",
    "stationRegion = [\"济南\", \"济南西\", \"济南东\", \"大明湖\"]\n",
    "\n",
    "\n",
    "Excelpath = \"data.xlsx\"  # Excel时刻表输入文件路径，如爬取的时刻表或是是复制来的\n",
    "TextPath = \"train.txt\"  # 游戏时刻表文件输出路径\n",
    "# 速度和编组以及类型映射关系,0为普速1为动车2为高速，后续修改\n",
    "species = {'新空调普快': ['120', 'LPPPPPP', 0], '新空调快速': ['120', 'LPPPPPP', 0],\n",
    "           '新空调特快': ['140', 'LPPPPPP', 0], '新空调直快': ['160', 'LPPPPPP', 0],\n",
    "           '动车': ['200', 'LPPL', 1], '城际': ['200', 'LPPL', 1], '高速': ['300', 'LPPLLPPL', 2]}\n",
    "species1 = {'K': ['120', 'LPPPPPP', 0], 'T': ['140', 'LPPPPPP', 0], 'Z': ['160', 'LPPPPPP', 0],\n",
    "            'D': ['200', 'LPPL', 1], 'C': ['200', 'LPPL', 1], 'G': ['300', 'LPPLLPPL', 2]}\n",
    "\n",
    "# 车站-编号,掉向,用时以及运行车辆种类映射关系\n",
    "# 图片左(0)右(1)侧线路key值相同则掉向,\n",
    "# 国铁车辆行走左侧,2为数据为左侧股道编号\n",
    "# [车站编号,车站所在侧(0为左侧),车辆进场股道,车辆离场行走股道,到达中心车站所用时间]\n",
    "station = {\n",
    "    '京济联络线济南方向': ['b', 0, 2, 1, 8], '济郑高速长清方向': ['c', 0, 2, 1, 7],\n",
    "    '济南西站': ['d', -1, 0, 0, 7], '济南西动车所': ['e', -1, 0, 0, 30],\n",
    "    '京沪高速德州东方向': ['f', 1, '1', '2', 6, ], '京沪高速泰安站方向': ['a', 0, 2, 1, 7, ],\n",
    "    '石济客专齐河方向': ['g', 1, 1, 2, 6], '石济客专济南东方向': ['h', 1, 1, 2, 6, ],\n",
    "}\n",
    "\n",
    "# 线路和车站关系，主要用于从车站-值获取线路-键\n",
    "route1 = {\n",
    "    '济南西动车所': [\"济南西\"], '京济联络线济南方向': [\"济南\"], '济郑高速长清方向': [\"长清\"],\n",
    "    '京沪高速泰安站方向': [\"泰安\", \"曲阜东\", \"滕州东\", \"枣庄\", \"徐州东\", \"宿州东\", \"蚌埠南\", \"南京南\"],\n",
    "    '京沪高速德州东方向': [\"北京南\", \"天津西\", \"天津\", \"沧州西\", \"德州东\"],\n",
    "    '石济客专齐河方向': [\"齐河\", \"禹城东\", \"平原东\"], '石济客专济南东方向': [\"济南东\"]}\n",
    "# 两个合在一起写太难看\n",
    "\n",
    "\n",
    "# 车型关系--待完善，后续待交路表部分完成后会使用对应映射而不是上面的简单类型映射\n",
    "marshalling = {'CHR380BL': [], 'CHR380B': [], 'CHR380B重连': []}\n",
    "\n",
    "\n",
    "ThisStation = '济南西站'\n",
    "TotalPlat = 18\n",
    "# track = {}\n",
    "\n",
    "\n",
    "# 模拟器信息\n",
    "desired_caps = {\n",
    "    'platformName': 'Android',  # 被测手机是安卓\n",
    "    'platformVersion': '9',  # 手机安卓版本\n",
    "    'deviceName': 'emulator-5554:5555',  # 设备名和映射连接端口，安卓手机可以随意填写\n",
    "\n",
    "    'appPackage': 'com.lltskb.lltskb',  # 启动APP Package名称\n",
    "    'appActivity': '.ui.splash.SplashActivity',  # 启动Activity名称\n",
    "\n",
    "    'unicodeKeyboard': True,  # 使用自带输入法，输入中文时填True\n",
    "    'resetKeyboard': True,  # 执行完程序恢复原来输入法\n",
    "    'noReset': True,       # 不要重置App\n",
    "    'newCommandTimeout': 6000,\n",
    "    'automationName': 'UiAutomator2'\n",
    "}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 第二部分，appnium自动化爬取路路通上时刻表信息\n",
    "### 内容\n",
    "这部分主要有三个函数，分别是按照车站获取车次信息，按照车次查询--检票口查询获取该车次所有信息，以及车站大屏查询  \n",
    "数据均来自路路通，界面信息所见即所得  \n",
    "1. 车站信息  \n",
    "    1.1 这部分较为简单，爬取时信息较为规整。始发终到车站信息等界面上每一个格子都对应一个数据，按照正常顺序获取处理即可\n",
    "    1.2 滑动页面并获取数据，数据按照一行八个获取，以csv样式写入txt\n",
    "2. 车次信息--检票样式查询  \n",
    "    2.1 这部分难度就大很多，主要难点在于目前来看必须多次重复点击查询按钮  \n",
    "    2.2 同样滑动页面并获取数据，数据按照以车站核心一行四个数据，同样以csv样式写入文本txt\n",
    "3. 车站大屏信息  \n",
    "    3.1 由于网络问题应该相当长的时间内无法完成，主要目的是为了获取如汉口站，郑州站这类检票口信息不定导致无法离线查询的车站  \n",
    "\n",
    "### 效果\n",
    "1. 如果不需要做进路映射和股道映射，单纯通过车站信息爬取已经足够使用，已经有完善的实发终到，到时开时信息\n",
    "2. 如果需要做进路映射和股道映射且较为复杂，可以用后面车次信息部分来获取这部分信息，但是时长较大\n",
    "\n",
    "### 实现原理  \n",
    "1. appnium按流程进入页面并得到文本数据  \n",
    "2. 死循环获取页面数据，直到页面数据不在变化或者由于数据过少而无法变化直接退出  \n",
    "3. 使用set来收集数据以优化重复数据  \n",
    "\n",
    "### 待完善/优化的部分\n",
    "<del>目标更换 1. 优化车次信息查询逻辑，改善输入格式，进行多轮查询时可以更好的数据结果，为之后按照车次来部署整条线路做准备</del>  \n",
    "1. 优化车次信息查询效率，尽可能少重复点选\n",
    "2. 加入点击交路表并获取相关信息的部分，使得立折车次自动合并和正确映射编组成为可能\n",
    "3. 看看能不能使用车站大屏功能，可以获取停站股道不定的车站检票口停站信息  \n",
    "4. .......\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 每一个车站的所有停车信息\n",
    "\n",
    "def StationTrainList(driver: WebDriver, action: TouchAction, station: str):\n",
    "    time.sleep(8)  # 等待开屏广告\n",
    "    stationText = driver.find_element(\n",
    "        AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text(\"车站\")')\n",
    "\n",
    "    action.tap(stationText).perform()  # 点击车站选项的位置\n",
    "\n",
    "    stationChk = driver.find_element(\n",
    "        AppiumBy.ID, \"com.lltskb.lltskb:id/layout_station\")  # 车站选择部分\n",
    "    action.tap(stationChk).perform()  # 进入车站选择\n",
    "    time.sleep(1)\n",
    "    stationSelect = driver.find_element(\n",
    "        AppiumBy.ID, \"com.lltskb.lltskb:id/edit_input\")  # 车站选择输入框\n",
    "    stationSelect.send_keys(station)  # 输入\n",
    "\n",
    "    stationFind = driver.find_elements(\n",
    "        AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text(\"{0}\")'.format(station))\n",
    "    action.tap(stationFind[1]).perform()  # 输入车站名并确认，返回之前的页面\n",
    "\n",
    "    searchbtn = driver.find_element(\n",
    "        AppiumBy.ID, \"com.lltskb.lltskb:id/querycz_btn\")  # 搜索按钮\n",
    "    action.tap(searchbtn).perform()  # 点选进行查找\n",
    "    time.sleep(2)\n",
    "    # 当前页面可视的车次部分\n",
    "    pt = driver.find_element(AppiumBy.CLASS_NAME, 'android.widget.ListView')\n",
    "    x1 = pt.size['width'] * 0.5\n",
    "    y1 = pt.size['height']  # 获取页面宽高,方便后续滑动截取车次信息\n",
    "\n",
    "    f = open(file=\"{0}Info.txt\".format(station), mode=\"w\", encoding=\"utf-8\")\n",
    "    # 设置为utf8，防止如复兴号，静音动车等符号导致报错\n",
    "    settext = set()  # 使用set尽可能减少重复值\n",
    "\n",
    "    while True:  # 滑动截取文字\n",
    "        # 所有有文本的元素的集合，均为当前可见部分\n",
    "        pageTrainList = driver.find_elements(\n",
    "            AppiumBy.CLASS_NAME, 'android.widget.TextView')\n",
    "        beforeSwipe = driver.page_source\n",
    "\n",
    "        text = \"\"\n",
    "        for i in range(0, len(pageTrainList)-9):  # 排除表头信息，仅保留车次部分\n",
    "            text = text+pageTrainList[i+9].text+\",\"  # 获取整体信息\n",
    "            if (i+9) % 8 == 0:\n",
    "\n",
    "                settext.add(text+\"\\n\")\n",
    "                text = \"\"\n",
    "\n",
    "        try:\n",
    "            # 这里模拟器的分辨率是1920*1080的手机屏幕，不一样的需要改参数\n",
    "            # 如果内容少于一屏幕会直接报错，故使用try\n",
    "            action.long_press(\n",
    "                x=540, y=1800, duration=200).move_to(x=540, y=100).release()\n",
    "            action.perform()  # 在中间滑动，选取新的车次信息\n",
    "        # 如果滑动前后的元素相同，则表示已经到底了\n",
    "        except Exception:  # 少于一屏幕就表示已经完成了\n",
    "            break\n",
    "        else:  # 前后的页面元素一样\n",
    "            if driver.page_source == beforeSwipe:\n",
    "                break\n",
    "    for it in settext:\n",
    "        f.write(it)\n",
    "    f.close()\n",
    "\n",
    "    return\n",
    "\n",
    "# 一组一列车的所有停站信息\n",
    "\n",
    "\n",
    "def TrainStationList(driver: WebDriver, action: TouchAction, trainli: list[str], fn=\"appniumTrainInfo.txt\"):\n",
    "    time.sleep(8)  # 等待开屏广告\n",
    "    stationText = driver.find_element(\n",
    "        AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text(\"车次\")')\n",
    "    action.tap(stationText).perform()  # 点击车次选项的位置\n",
    "\n",
    "    stationchk = driver.find_element(\n",
    "        AppiumBy.ID, \"com.lltskb.lltskb:id/chk_jpk\")\n",
    "    action.tap(stationchk).perform()  # 点选查找检票口选项\n",
    "    # 设置为utf8，防止如复兴号，静音动车等符号导致报错\n",
    "    f = open(file=fn, mode=\"w\", encoding=\"utf-8\")\n",
    "    for train in trainli:\n",
    "        settext = set()  # 使用set尽可能减少重复值\n",
    "        btnClicked = []\n",
    "        trainSelect = driver.find_element(\n",
    "            AppiumBy.ID, \"com.lltskb.lltskb:id/edit_train\")\n",
    "        trainSelect.clear()  # 清空内容防止重输\n",
    "        trainSelect.send_keys(train)  # 输入查找车次\n",
    "\n",
    "        searchbtn = driver.find_element(\n",
    "            AppiumBy.ID, \"com.lltskb.lltskb:id/querycc_btn\")  # 搜索按钮\n",
    "        action.tap(searchbtn).perform()  # 点选进行查找\n",
    "\n",
    "        text = \"\"\n",
    "        while True:  # 进入页面\n",
    "            searchbtns = driver.find_elements(\n",
    "                AppiumBy.ID, \"com.lltskb.lltskb:id/btn_query\")  # 搜索按钮\n",
    "            beforeSwipe = driver.page_source\n",
    "\n",
    "            for btn in searchbtns:  # 当前页面上所有“查询”按钮，目前除了全部重复点一遍没有好办法\n",
    "                # if btn.id in btnClicked:\n",
    "                #     continue\n",
    "                action.tap(btn).perform()  # 点选所有按钮进行查找\n",
    "                btnClicked.append(btn.id)  # 只能用列表等来标记\n",
    "\n",
    "            pageStationList = driver.find_elements(\n",
    "                AppiumBy.CLASS_NAME, 'android.widget.TextView')\n",
    "           # 获取页面所有信息\n",
    "            for i in range(1, len(pageStationList)):\n",
    "                # 停站，到时，开时，检票口共四个停站信息\n",
    "                text = text+pageStationList[i].text+\",\"\n",
    "                if i % 4 == 0:  # 每一个车站\n",
    "                    text = train+\",\"+text+\"\\n\"\n",
    "                    settext.add(text)\n",
    "                    text = \"\"\n",
    "            try:\n",
    "                # 这里模拟器的分辨率是1920*1080的手机屏幕\n",
    "                # 如果内容少于一屏幕会直接报错，故使用try\n",
    "                action.long_press(\n",
    "                    x=540, y=1700, duration=150).move_to(x=540, y=200).release()\n",
    "                action.perform()  # 在中间滑动，选取新的车次信息\n",
    "                time.sleep(0.5)\n",
    "            # 如果滑动前后的元素相同，则表示已经到底了\n",
    "            except Exception:  # 少于一屏幕就表示已经完成了\n",
    "                print(Exception.args)\n",
    "                break\n",
    "            else:  # 前后的页面元素一样\n",
    "                if driver.page_source == beforeSwipe:\n",
    "                    break\n",
    "\n",
    "        # 按下返回按键，进行下一次查找\n",
    "        retbtn = driver.find_element(\n",
    "            AppiumBy.ID, \"com.lltskb.lltskb:id/btn_back\")\n",
    "        action.tap(retbtn).perform()\n",
    "        # 替换两个汉字，方便后续操作\n",
    "        print(settext)\n",
    "        for it in settext:\n",
    "            it = it.replace(\"到\", \"\").replace(\"开\", \"\")\n",
    "            f.write(it)\n",
    "        f.flush()  # 直接刷新缓冲区写入\n",
    "\n",
    "    f.close()\n",
    "\n",
    "    return\n",
    "\n",
    "\n",
    "# 未完成，车站大屏查询\n",
    "\n",
    "\n",
    "def StationScreenList(driver: WebDriver, action: TouchAction, train: str):\n",
    "    stationText = driver.find_element(\n",
    "        AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text(\"车站\")')\n",
    "\n",
    "    action.tap(stationText).perform()  # 点击车站选项的位置\n",
    "\n",
    "    stationChk = driver.find_element(\n",
    "        AppiumBy.ID, \"com.lltskb.lltskb:id/btn_big_screen\")  # 车站大屏选择\n",
    "    action.tap(stationChk).perform()  # 车站大屏内容部分\n",
    "\n",
    "    return\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b> 爬取部分无需多次运行 </b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps)\n",
    "# 设置缺省等待时间\n",
    "driver.implicitly_wait(10)\n",
    "action = TouchAction(driver)  # 点选操作的对象\n",
    "\n",
    "targetSt = \"济南西\"\n",
    "# 济南Info是之前爬取的济南地区车站信息\n",
    "stationFrame = pandas.read_csv(filepath_or_buffer=\"济南Info1.txt\",\n",
    "                               sep=\",\", encoding=\"utf-8\")\n",
    "stationFrame.columns = [\"车次\", \"drop1\", \"停站\",\n",
    "                        \"到时\", \"开时\", \"始发站\", \"终到站\", \"列车类型\", \"drop2\"]\n",
    "# 删去重复列和空列\n",
    "stationFrame.drop_duplicates(inplace=True)\n",
    "stationFrame.drop(columns=[\"drop1\", \"drop2\"], inplace=True)\n",
    "\n",
    "# 获取所有济南西站的车次信息\n",
    "jnzlist = stationFrame[stationFrame[\"停站\"] == targetSt][\"车次\"].to_list()\n",
    "TrainStationList(driver, action, jnzlist, \"jinanxitest2.txt\")\n",
    "\n",
    "time.sleep(3)\n",
    "# driver.close()\n",
    "driver.quit()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 第三部分，处理获取到的数据并生成游戏样式的时刻表\n",
    "这部分最为重要，按照游戏中时刻表格式分为若干部分  \n",
    "车辆信息部分，如车次编号，速度等级等信息  \n",
    "停站信息部分，有四部分按顺序分别为 停站编号，停站股道，停站时刻，停站时间。这四部分会单独拆开分别生成  \n",
    "例： G7 COMMUTER 300 LPPLLPPL X1 : f#1#11:17:00#0 d#6#11:23:00#3 a#1#11:33:00#0  \n",
    "\n",
    "内容|完成度|实现原理|近期改良计划|远期优化计划\n",
    "---|---|---|---|---\n",
    "车辆信息|较为完善|目前为仅按照车次编号和类型生成|近期暂无|完善交路表爬取以修改速度级和编组信息立折车次信息完善\n",
    "停站信息-停站编号|完善|依照游戏自身设置直接映射|暂无|暂无\n",
    "停站信息-停站股道|较为完善|selenium按照web自动化12306查询获取 <br>appnium部分按照车次信息查询--检票口获得|完善车次查询部分以获得<br>停站检票口信息|使用车站大屏来获得无法离线查询的车站信息\n",
    "停站信息-停站时刻|完善|按照到站时间和出发时间推算|近期暂无|加上线路所映射行车进路部分\n",
    "停站信息-停站时间|完善|按照到站时间和出发时间计算|近期暂无|立折车次停站时间等按照交路表推算\n",
    "整体上|较为完善|使用apply(lambda) 来进列映射整处理而不是按照行单个处理|暂无  \n",
    "\n",
    "### 第三部分第〇小部分--处理用函数定义\n",
    "#### 内容  \n",
    "目前按照上面输入为上面自动化爬取获取到csv样式的txt文件，输出为按照车站为单位的数据  \n",
    "1. 首先去除空行和重复值等基本操作进行数据提纯   \n",
    "2. 在统一处理如得到部分信息如停站时间和始发终到的处理，  \n",
    "3. 可以获得中心车站的停站时间信息等  \n",
    "4. 同时预先定义需要使用的处理函数  \n",
    "\n",
    "#### 实现原理  \n",
    "1. 整体上以处理dataframe为主，文件→dataframe→文本\n",
    "2. 大量使用.apply相应处理函数来处理列，尽可能少使用匿名函数来增加可读性\n",
    "3. 由文件读出的大表来初步处理groupby 再对想到较小的表来处理\n",
    "4.\n",
    "\n",
    "\n",
    "#### 待完善/优化的部分\n",
    "近期暂无计划，远期计划加入进路映射以使用线路所来优化自动化  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 数据处理用\n",
    "import pandas\n",
    "import numpy\n",
    "import time\n",
    "import datetime\n",
    "import random\n",
    "import itertools\n",
    "\n",
    "def dealTime(datafr: pandas.DataFrame) -> pandas.DataFrame:\n",
    "    # 对于始发车和终到车进行时间处理，默认相同\n",
    "    for idx, row in datafr.iterrows():\n",
    "        if row[\"到时\"] == \"--:--\":\n",
    "            row[\"到时\"] = row[\"开时\"]\n",
    "        elif row[\"开时\"] == \"--:--\":\n",
    "            row[\"开时\"] = row[\"到时\"]\n",
    "    # 计算停站时间\n",
    "    datafr[\"到时\"] = pandas.to_datetime(datafr[\"到时\"])\n",
    "    datafr[\"开时\"] = pandas.to_datetime(datafr[\"开时\"])\n",
    "    # 实发终到设置停站时间\n",
    "    # pd[\"停站时间\"].replace(0,10,inplace=True)\n",
    "\n",
    "    return datafr\n",
    "\n",
    "\n",
    "def stopStTime(stopTime: pandas.Timedelta) -> int:\n",
    "    a = int(stopTime.seconds/60)  # 处理为数值类型的分钟时间\n",
    "    # 始发或是终到的时间是随机的\n",
    "    return a if a > 0 else random.randint(10, 30)\n",
    "\n",
    "\n",
    "def ModtimeStr(switchTime: pandas.Timestamp) -> str:\n",
    "    # 处理为仅有时分秒样式的字符串格式的时间\n",
    "    return switchTime.strftime(\"%H:%M:%S\")\n",
    "\n",
    "\n",
    "def speMarType(traincode: str) -> str:  # 目前获取编组信息的方式\n",
    "    til = species1[traincode[0]]  # 处理获取车辆信息字符串部分\n",
    "    return \"{cod} COMMUTER {speed} {mar} X1\".format(cod=traincode, speed=til[0], mar=til[1])\n",
    "\n",
    "\n",
    "def checkin(entrance: str) -> int:\n",
    "    # 终到车\n",
    "    res = 0\n",
    "    if entrance == \"无检票口信息\":\n",
    "        res = 0\n",
    "    else:  # 格式类似检票口15A，替换掉其他字样\n",
    "        res = int(entrance.strip(\"检票口AB\"))\n",
    "    return res\n",
    "\n",
    "\n",
    "def prevnextST(stopSt: list[str], stIdx: int) -> list[str]:\n",
    "    # 返回结果，获取目标车站的前后车站\n",
    "    res = [\"\", \"\"]\n",
    "    res1 = [\"\", \"\"]\n",
    "    if stIdx == 0:  # 始发车视为始发车站和下一站\n",
    "        res = [stopSt[0], stopSt[1]]\n",
    "    elif stIdx == len(stopSt)-1:  # 终到车视为前一站和终到站\n",
    "        res = [stopSt[len(stopSt)-2], stopSt[len(stopSt)-1]]\n",
    "    else:  # 中间站\n",
    "        res = [stopSt[stIdx-1], stopSt[stIdx+1]]\n",
    "    # 判断并修改进路\n",
    "    for k, v in route1.items():\n",
    "        # 把车站名映射为线路\n",
    "        if res[0] in v:\n",
    "            res1[0] = k\n",
    "        elif res[1] in v:\n",
    "            res1[1] = k\n",
    "    return res1\n",
    "\n",
    "\n",
    "def arrLeaTime(t1: pandas.Timestamp, st: list[str], mark: int) -> str:\n",
    "    # 选取是上一站还是下一站\n",
    "    tarst = st[mark]\n",
    "    useTime = station[tarst][4]\n",
    "    if mark == 0:  # 进场减时间\n",
    "        res = t1-datetime.timedelta(minutes=useTime)\n",
    "    else:  # 离场加时间\n",
    "        res = t1+datetime.timedelta(minutes=useTime)\n",
    "    # 返回时分秒格式的字符串\n",
    "    return res.strftime(\"%H:%M:%S\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 第三部分第一小部分--爬取信息整体初步处理 \n",
    "#### 内容\n",
    "1. 初步整理上面按照<b>车次检票口查询</b>到的数据\n",
    "2. 按照车次-进站时间整理后进行分组，按照车次聚合为数据框\n",
    "3. 获取上下站的信息等，完成初步处理\n",
    "\n",
    "#### 实现原理\n",
    "缺省值填充等数据清洗操作  \n",
    "重分组来重新构造数据特征  \n",
    "#### 待完善/优化的部分\n",
    "暂无"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 手动建立映射关系,别的方法不会\n",
    "\n",
    "\n",
    "def processTrains(fn: str, targetSt: str) -> pandas.DataFrame:\n",
    "    # 处理产生初步所需信息的数据框\n",
    "    trainsFrame = pandas.read_csv(filepath_or_buffer=fn,\n",
    "                                  sep=\",\", encoding=\"utf8\", header=None)\n",
    "    trainsFrame.columns = [\"车次\",  \"停站\", \"到时\", \"开时\", \"检票口\", \"drop1\"]\n",
    "    trainsFrame.drop(columns=[\"drop1\"], inplace=True)\n",
    "\n",
    "    trainsFrame = dealTime(trainsFrame)  # 初步处理时间\n",
    "    trainsFrame.sort_values(by=[\"车次\", \"到时\"], inplace=True)  # 归类排序\n",
    "    # 整理为分钟样式的停站时间\n",
    "    #trainsFrame[\"停站时间\"] = trainsFrame[\"停站时间\"].apply(lambda x: stopStTime(x))\n",
    "    trainsFrame.reset_index(drop=True, inplace=True)\n",
    "    # 按照车次分组为列表，同时车次不作为索引以便于操作\n",
    "    tfgp = trainsFrame.groupby(by=[\"车次\"], as_index=False).agg(list)\n",
    "    # 获取目标停站的索引值，设置为一个辅助列\n",
    "    tfgp[\"auxIdx\"] = tfgp['停站'].apply(lambda x: x.index(targetSt))\n",
    "\n",
    "    # 获取前后停站以得到更好的交路映射\n",
    "    tfgp[\"目标前后站\"] = tfgp.apply(\n",
    "        lambda x: prevnextST(x[\"停站\"], x[\"auxIdx\"]), axis=1)\n",
    "    # 获取其他信息\n",
    "    tfgp[\"目标站到时\"] = tfgp.apply(lambda x: x[\"到时\"][x[\"auxIdx\"]], axis=1)\n",
    "    tfgp[\"目标站开时\"] = tfgp.apply(lambda x: x[\"开时\"][x[\"auxIdx\"]], axis=1)\n",
    "    tfgp[\"目标站检票口\"] = tfgp.apply(lambda x: x[\"检票口\"][x[\"auxIdx\"]], axis=1)\n",
    "\n",
    "    # 提取需要的信息为新的dataframe\n",
    "    trainInfoNeed = tfgp[[\"车次\", \"目标站到时\", \"目标站开时\", \"目标站检票口\", \"目标前后站\"]]\n",
    "    trainInfoNeed.columns = [\"车次\", \"到时\", \"开时\", \"检票口\", \"前后站\"]\n",
    "    # print(trainInfoNeed.head())\n",
    "\n",
    "    return trainInfoNeed\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  第三部分第二小部分 -- 处理生成最终dataframe样式的数据 \n",
    "\n",
    "#### 内容\n",
    "按照上面的函数处理数据，最终处理生成有进(离)场时间/股道 等信息的dataframe,之后生成字符串导出    \n",
    "这里分为两部分，车辆信息的trainframe和路径时刻表的stationframe,    \n",
    " 'D8176 COMMUTER 200 LPPL X1 : b#2#18:08:00#0 d#0#18:16:00#14 e#0#18:36:00#0 ',  \n",
    "<b>如果不做股道映射直接在等号后面写0即可，或是使用如随机数做映射等操作</b>  \n",
    "\n",
    "#### 实现原理\n",
    "> 1. generateTrainInfo 即为文字版时刻表分号前的内容，生成车次信息部分  \n",
    ">> 1.1. 考虑到内容只需做车次编号，速度等级和编组信息则按照车次首位ktz生成    \n",
    ">> 2.1. 编组信息尚未准备开始   \n",
    "> 2. generateArriveLeave 即为文字版时刻表分号后的内容，生成进场和离场信息部分  \n",
    ">> 2.1. 基于停站的上一站完善进路的刻画\n",
    ">> 2.2. 本身进场和离场处理基本类似，这里使用一个mark来进行区分\n",
    "> 3. generateMainSt 即为区域核心车站生成部分  \n",
    ">> 3.1. 绝大多数信息已经在前面处理完成，这里主要是处理格式问题  \n",
    ">> 3.2. 车站名称的填充放在最后，防止由于数据全空导致填充失败  \n",
    "\n",
    "#### 改进计划  \n",
    "1. 对于分场式车站采用进场线路--对应线路所来映射，加入线路所映射部分\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "def generateRoute(dataser: pandas.Series):\n",
    "    # 进行中，对于分场式车站采用线路所映射进路\n",
    "    s1 = \"黄河南线路所\"\n",
    "    s2 = \"大漠刘线路所\"\n",
    "    return None\n",
    "#\n",
    "\n",
    "def generateArriveLeave(datafr: pandas.DataFrame, mark: int) -> pandas.DataFrame:\n",
    "    # 生成进场/离场信息\n",
    "    # mark=0为进场，mark=1为离场\n",
    "    ALStDF = pandas.DataFrame(columns=['车站名称', '股道', '到达时间', '停站时间'])  # 进场信息\n",
    "    # 按照信息来映射进场/离场线路\n",
    "    ALStDF['车站名称'] = datafr[\"前后站\"].apply(lambda x: x[mark])\n",
    "    # 按照信息来映射进场/离场股道\n",
    "    ALStDF['股道'] = ALStDF['车站名称'].apply(lambda x: station[x][2+mark])\n",
    "    # ALStDF['股道'] = 0 #不做股道映射则直接填0即可\n",
    "    # 时间作差\n",
    "    ALStDF['到达时间'] = datafr.apply(\n",
    "        lambda x: arrLeaTime(x[\"到时\"], x[\"前后站\"], mark), axis=1)\n",
    "    # 进程默认经过不停车\n",
    "    ALStDF['停站时间'] = 0\n",
    "\n",
    "    return ALStDF\n",
    "\n",
    "\n",
    "def generateMainSt(datafr: pandas.DataFrame):\n",
    "    # 生成中心停站信息\n",
    "    MainStDF = pandas.DataFrame(columns=['车站名称', '股道', '到达时间', '停站时间'])  # 进场信息\n",
    "\n",
    "    # 计算停站时长，得到整形类型的时长，非浮点数没有.0便于写入文件\n",
    "    MainStDF[\"停站时间\"] = (datafr[\"开时\"]-datafr[\"到时\"]\n",
    "                        ).apply(lambda x: stopStTime(x))\n",
    "    # 计算并规整进场离场和停站时间，将三个到时改为游戏时分秒格式 hh:mm:ss\n",
    "    MainStDF[\"到达时间\"] = datafr[\"到时\"].apply(lambda x: ModtimeStr(x))\n",
    "    # 按照相应格式处理车站检票口为对应形式\n",
    "    MainStDF[\"股道\"] = datafr[\"检票口\"].apply(lambda x: checkin(x))\n",
    "    # 填充车站为中心车站\n",
    "    MainStDF[\"车站名称\"].fillna(value=\"济南西\", inplace=True)\n",
    "    return MainStDF\n",
    "\n",
    "\n",
    "def generateTrainInfo(dataser: pandas.Series) -> pandas.Series:\n",
    "    # 生成车次信息\n",
    "    res = dataser.apply(lambda x: speMarType(x))\n",
    "    print(res.head())\n",
    "    return res\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 第四部分 -- 生成对应样式的字符串并写入文件\n",
    "\n",
    "### 内容  \n",
    "1. 整合之前的数据来生成符合样式的dataframe数据合集  \n",
    "2. 生成csv长字符串并按行分割为一个个的端数据  \n",
    "3. 最终连接处理生成结果，不替换代号便于后续检查    \n",
    "\n",
    "### 实现原理\n",
    "整合数据和字符串处理\n",
    "\n",
    "### 改进计划  \n",
    "暂无"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def generateStr(fn: str) -> list:\n",
    "    # 生成游戏样式的字符串\n",
    "    f = open(file=fn, encoding=\"utf8\", mode=\"w\")\n",
    "    res = []\n",
    "    for i in range(0, len(arriveStDF)):\n",
    "        # format为对应顺序格式\n",
    "        tf = \"{train} : {arrive} {stop} {leave}\".format(\n",
    "            train=trainstr[i], arrive=arrivestr[i], stop=stopstr[i], leave=leavestr[i])\n",
    "        print(tf)\n",
    "        res.append(tf)  # 将结果写入文件\n",
    "        f.write(tf+\"\\n\")\n",
    "\n",
    "    f.close()\n",
    "    return res\n",
    "\n",
    "# 实际运行部分\n",
    "targetSt = \"济南西\"\n",
    "at = processTrains(\"jinanxitest1.txt\", targetSt)\n",
    "# 生成dataframe格式的数据\n",
    "arriveStDF = generateArriveLeave(at, mark=0)\n",
    "leaveStDF = generateArriveLeave(at, mark=1)\n",
    "stopStDF = generateMainSt(at)\n",
    "trainDF = generateTrainInfo(at[\"车次\"])\n",
    "\n",
    "# 按照四部分生成需求的字符格式，使用to_csv函数来生成\n",
    "arrivestr = arriveStDF.to_csv(\n",
    "    sep=\"#\", header=False, index=False).split(sep=\"\\r\\n\")\n",
    "leavestr = leaveStDF.to_csv(sep=\"#\", header=False,\n",
    "                            index=False).split(sep=\"\\r\\n\")\n",
    "stopstr = stopStDF.to_csv(sep=\"#\", header=False, index=False).split(sep=\"\\r\\n\")\n",
    "trainstr = trainDF.to_csv(header=False, index=False).split(sep=\"\\r\\n\")\n",
    "\n",
    "generateStr(\"appnium{0}测试.txt\".format(targetSt))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "-------  \n",
    "\n",
    "## 第五部分，可视化用于分析分布如高峰期和主要类型"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# 对于该车站群组（地区）时间段内的所有车次信息\n",
    "\n",
    "\n",
    "def getStationProid(datafr: pandas.DataFrame, at: int) -> dict:\n",
    "    # 预先分类，按照小时\n",
    "    datagruop = datafr.groupby([\"到时-小时\", \"停站\", \"列车类型\"])\n",
    "    # print(datagruop.groups)\n",
    "    regionInfo = {}\n",
    "\n",
    "    # 对于每个车站的每种类型的列车\n",
    "    for sr in stationRegion:\n",
    "        # 好像只能分开写两层了\n",
    "        regionInfo[sr+\"trainCode\"] = []\n",
    "        regionInfo[sr+\"trainCount\"] = []\n",
    "        for tt in trainType:\n",
    "            try:  # 查找符合条件的群组\n",
    "                i = datagruop.get_group((at, sr, tt))\n",
    "\n",
    "            except KeyError:  # 索引找不到就是0\n",
    "                regionInfo[sr+\"trainCount\"].append(0)\n",
    "                # regionInfo[\"trainCode\"].append(None)\n",
    "            else:  # 如果没问题，添加回去长度\n",
    "\n",
    "                regionInfo[sr+\"trainCode\"].append(i[\"车次\"].tolist())\n",
    "                regionInfo[sr+\"trainCount\"].append(len(i))\n",
    "        regionInfo[sr+\"trainCode\"] = sum(regionInfo[sr+\"trainCode\"], [])\n",
    "    # 解包嵌套的列表\n",
    "    # print(regionInfo)\n",
    "\n",
    "    return regionInfo\n",
    "\n",
    "\n",
    "def getTrainProid(datafr: pandas.DataFrame, at: int) -> dict:\n",
    "    # 预先分类，数据分析用\n",
    "    datagruop = datafr.groupby([\"到时-小时\", \"停站\", \"列车类型\"])\n",
    "    # print(datagruop.groups)\n",
    "    regionInfo = {}\n",
    "\n",
    "    # 对于每个车站的每种类型的列车\n",
    "    for tt in trainType:\n",
    "        # 好像只能分开写两层了\n",
    "        regionInfo[tt+\"trainCode\"] = []\n",
    "        regionInfo[tt+\"stationName\"] = []\n",
    "        regionInfo[tt+\"stationCount\"] = []\n",
    "        for st in stationRegion:\n",
    "            try:  # 查找符合条件的群组\n",
    "                i = datagruop.get_group((at, st, tt))\n",
    "\n",
    "            except KeyError:  # 索引找不到就是0\n",
    "                regionInfo[tt+\"stationCount\"].append(0)\n",
    "                # regionInfo[\"trainCode\"].append(None)\n",
    "            else:  # 如果没问题，添加回去长度\n",
    "                regionInfo[tt+\"trainCode\"].append(i[\"车次\"].tolist())\n",
    "                regionInfo[tt+\"stationName\"].append(i[\"停站\"].tolist())\n",
    "                regionInfo[tt+\"stationCount\"].append(len(i))\n",
    "\n",
    "        regionInfo[tt+\"stationName\"] = sum(regionInfo[tt+\"stationName\"], [])\n",
    "    # 解包嵌套的列表\n",
    "    # print(regionInfo)\n",
    "\n",
    "    return regionInfo\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.2"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
